package com.jiayuan.cn.energy.test;

import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;

public class SeeTable {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

//// === EXAMPLE 1 ===
//        DataStream<Row> dataStream = env.fromElements(
//                Row.of("Alice", 12),
//                Row.of("Bob", 10),
//                Row.of("Alice", 100));
//
//        Table inputTable = tableEnv.fromChangelogStream(dataStream).as("name", "score");
//        inputTable.printSchema();
// prints:
// (
//  `name` STRING,
//  `score` INT
// )

//// === EXAMPLE 2 ===
//        DataStream<Row> dataStream = env.fromElements(
//                Row.of("Alice", 12),
//                Row.of("Bob", 10),
//                Row.of("Alice", 100));
//
//        DataStream<Row> processedStream = dataStream.process(new ProcessFunction<Row, Row>() {
//            @Override
//            public void processElement(Row row,
//                                       Context context,
//                                       Collector<Row> collector) {
//                collector.collect(row);
//            }
//        });
//
//        Table inputTable = tableEnv.fromChangelogStream(processedStream);
//        inputTable.printSchema();
// prints:
// (
//  `f0` RAW('org.apache.flink.types.Row', '...')
// )

// === EXAMPLE 3 ===
        DataStream<Row> dataStream = env.fromElements(
                Row.of("Alice", 12),
                Row.of("Bob", 10),
                Row.of("Alice", 100));

        RowTypeInfo rowTypeInfo = new RowTypeInfo(
                new TypeInformation<?>[] {Types.STRING, Types.INT},
                new String[] {"name", "score"});
        DataStream<Row> processedStream = dataStream.process(new ProcessFunction<Row, Row>() {
            @Override
            public void processElement(Row row,
                                       Context context,
                                       Collector<Row> collector) {
                collector.collect(row);
            }
        }).returns(rowTypeInfo);

        Table inputTable = tableEnv.fromChangelogStream(processedStream);
        inputTable.printSchema();
// prints:
// (
//  `name` STRING,
//  `score` INT
// )
    }
}
